package com.winning.platwebservice.service.Impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.winning.plat.framework.util.Des3;
import com.winning.platwebservice.controller.KafkaCollect;
import com.winning.platwebservice.dao.DataProcessingDao;
import com.winning.platwebservice.dao.SjyyxxDao;
import com.winning.platwebservice.entity.SjyyxxEntity;
import com.winning.platwebservice.service.IDataService;
import com.winning.platwebservice.utils.ValidateByJsonUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;

@Service
public class DataServiceImpl implements IDataService {

    private static final Logger logger = LogManager.getLogger(DataServiceImpl.class);


    @Autowired
    JdbcTemplate dbDataSource;

    @Autowired
    private SjyyxxDao sjyxxDao;

    @Autowired
    DatabaseMetaData md;

    @Autowired
    DataProcessingDao receiveDao;

    @Autowired
    private KafkaCollect kafkaCollect;

    @Autowired
    ValidateByJsonUtils validateByJsonUtils;


    @Value("${kafka.collect.enable}")
    private boolean enable;

    @Value("${sfjm}")
    private String sfjm;

    @Override
    @Transactional
    public String DataService(String params) {
        JSONObject resObj = new JSONObject();
        try {
            resObj.put("statuscode", "400");
            resObj.put("repMsg", "请求数据格式错误！");

            if ("1".equals(sfjm)) {  //如果参数加密，需要先解密
                try {
                    params = Des3.decode(params);
                } catch (Exception e) {
                    params = "";
                }
            }
            JSONObject reqObj = JSONObject.parseObject(params);
            JSONObject obHead = (JSONObject) reqObj.get("header");
            JSONArray arrBody = reqObj.getJSONArray("body");
//            JSONArray arrBody = (JSONArray) reqObj.get("body");

            String jgbm = obHead.get("orgId") == null ? "" : obHead.get("orgId").toString(); //机构编码
            String jgmc = obHead.get("orgName") == null ? "" : obHead.get("orgName").toString();
            String bm = obHead.get("dataSet") == null ? "" : obHead.get("dataSet").toString(); //表名

            //校验入参是否合法
            if (StringUtils.isEmpty(jgbm)) {
                resObj.put("statuscode", "400");
                resObj.put("repMsg", "数据不符合要求，节点orgId不可为空！");
                return resObj.toString();
            }

            if (StringUtils.isEmpty(jgmc)) {
                resObj.put("statuscode", "400");
                resObj.put("repMsg", "数据不符合要求，节点orgName不可为空！");
                return resObj.toString();
            }

            if (StringUtils.isEmpty(bm)) {
                resObj.put("statuscode", "400");
                resObj.put("repMsg", "数据不符合要求，节点dataSet不可为空！");
                return resObj.toString();
            }

            //得到主键列表
            Set<String> primaryKeys = getPrimaryKey(bm);


            for (int i = 0; i < arrBody.size(); i++) {
                JSONObject objBody1 = arrBody.getJSONObject(i);
                Map<String, String> stringStringMap = validateByJsonUtils.checkData(objBody1, bm);
                if (StringUtils.equals(stringStringMap.get("res"), "0")) {
                    resObj.put("statuscode", "400");
                    resObj.put("repMsg", stringStringMap.get("msg"));
                    return resObj.toString();
                }
                /*进行一次日期格式转换*/
                objBody1 = ValidateByJsonUtils.jsonDateConversion(objBody1);

                Map<String, Object> primaryKey = new LinkedHashMap();

                for (String str : primaryKeys) {
                    primaryKey.put(str, objBody1.get(str));
                }


                if (enable) {
                    Map sendData = new HashMap();
                    sendData.put("tableNmae", bm);
                    sendData.put("primaryKey", primaryKey);
                    sendData.put("data", objBody1);

                    kafkaCollect.producer(sendData);
                } else {
                    Integer s1 = receiveDao.selectDate(bm, primaryKey);
                    if (s1 > 0) {
                        receiveDao.deleteData(bm, primaryKey);
                    }
                    receiveDao.insertORupdate(bm, objBody1);
                }
            }


            resObj.put("statuscode", "200");
            resObj.put("repMsg", "操作成功！");
            return resObj.toString();


        } catch (SQLException e) {
            logger.error("SQL执行异常{}", e);
        }
        return resObj.toString();
    }

    /**
     * 根据表名得到表主键
     *
     * @param tableNmae
     * @return Set<String>
     */
    private synchronized Set<String> getPrimaryKey(String tableNmae) throws SQLException {

        ResultSet pkRset = null;
        try {
            pkRset = md.getPrimaryKeys(null, null, tableNmae);
            Set<String> pkName = new HashSet<String>();
            // 获取表主键名称列表
            while (pkRset.next()) {
                pkName.add(pkRset.getString("COLUMN_NAME"));
            }

            return pkName;
        } catch (Exception e) {
            logger.error("获取{}主键列表失败", tableNmae, e);
            throw new RuntimeException(e);
        } finally {
            if (pkRset != null) {
                pkRset.close();
            }
        }

    }

    @Override
    public List<SjyyxxEntity> getSjyyxx() {
        return sjyxxDao.getSjyyxx();
    }


}
